<?php


namespace coco\queue;

use coco\App;
use coco\Exception;
use coco\Log;
use coco\Tree;
use coco\Db;
use coco\Config;

final class Worker
{

    private $terminate = false; //是否中断
    private $queue;
    private static $last_free = 0;

    public function __construct($queue)
    {
        $this->queue = $queue;

        cli_set_process_title('coco queue ' . $this->queue . ' worker start_file=' . dirname(App::getInstance()->app_path));
        $this->initSignal();
        $this->log('start:' . posix_getpid() . ' ' . posix_getppid());
    }

    public function __destruct()
    {
        $this->log('stop:' . posix_getpid() . ' ' . posix_getppid());
    }

    public function run()
    {
        while (1) {
            //20分钟释放一次
            if (time() - self::$last_free > 1200 && self::$last_free > 0) {
                Db::free();
                self::$last_free = time();
            }
            pcntl_signal_dispatch();
            coco_sleep(0.01); //防止cpu长红
            if ($this->terminate) {
                break;
            }
            try {
                $job = Job::pop($this->queue);
                if (empty($job)) {
                    sleep(1);
                    continue;
                }
                try {
                    list($obj, $method) = $job['caller'];
                    $ref_class = new \ReflectionClass($obj);
                    if (!$ref_class->hasMethod($method) && !$ref_class->hasMethod('__call')) {
                        throw new Exception('非法操作:' . $method);
                    }
                    $try_times = $job['try_times'] ?? 0;
                    $jober = new Job($this->queue, $job['param'] ?? [], $try_times);
                    Tree::getInstance()->set(Job::class, $jober);
                    //如果存在构造方法
                    if ($constructor = $ref_class->getConstructor()) {
                        $argv = getMethodParams($constructor);
                        $class = $ref_class->newInstanceArgs($argv);
                    } else {
                        $class = $ref_class->newInstance();
                    }
                    if ($ref_class->hasMethod($method)) {
                        $ref_method = $ref_class->getMethod($method);
                        $argv = getMethodParams($ref_method);
                    } else {
                        $argv = [$method, []];
                    }
                    //返回false立即重试
                    if (false === call_user_func([$class, $method], ...$argv)) {
                        if ($try_times < 10) {
                            $job['try_times'] = $try_times + 1;
                            Job::rePush($this->queue, $job);
                            continue;
                        }
                        $fail = [
                            'job' => $job,
                        ];
                        Job::fail($this->queue, $fail);
                        continue;
                    }
                    Job::complete($this->queue, $job);
                } catch (\Throwable $e) {
                    $exception = [
                        'job' => $job,
                        'exception' => $e->getMessage() . ' ' . $e->getFile() . ':' . $e->getLine(),
                        'trace' => $e->getTraceAsString()
                    ];
                    Job::exception($this->queue, $exception);
                }
            } catch (\Throwable $e) {
                //开发异常忽略
            }
        }
    }

    private function log($msg)
    {
        Log::getInstanceByKey('queue')->write('queue ' . $this->queue . ' worker :' . $msg);
    }


    private function initSignal()
    {
        pcntl_signal(SIGUSR1, SIG_DFL);
        pcntl_signal(SIGCHLD, SIG_DFL);
        pcntl_signal(SIGTERM, [$this, 'signalHandler']);
        pcntl_signal(SIGHUP, [$this, 'signalHandler']);
        pcntl_signal(SIGQUIT, [$this, 'signalHandler']);
    }

    //信号处理函数
    private function signalHandler($signo)
    {
        switch ($signo) {
                //中断进程
            case SIGTERM:
            case SIGHUP:
            case SIGQUIT:
                $this->terminate = true;
                break;
            default:
        }
    }
}
